package com.xn.monitor.web.service;

import com.xn.monitor.common.SerializeUtil;
import com.xn.monitor.core.ApplicationMonitorRegister;
import com.xn.monitor.entity.ServerEventInfo;
import com.xn.monitor.web.notify.NotifyService;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Collection;

/**
 * zookeeper事件监听器
 * 监听/monitor/state节点，被监听应用自身停止时修改state节点，因此监听器间可以接监测到应用的启动、停止
 *
 * Created by yangshoujun on 2017/7/23 上午12:06.
 */
@Component
public class ZkStateSubscriptor extends java.util.Observable implements BeanFactoryPostProcessor {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    ZkClient zkClient = new ZkClient(PropertyUtils.get("zk.url"), 5000, 5000);

    private Collection<NotifyService> notifyServices;

    @Autowired
    private ThreadPoolTaskExecutor executor;

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) {

        if(!zkClient.exists(ApplicationMonitorRegister.STATE_PATH)) {
            zkClient.create(ApplicationMonitorRegister.STATE_PATH, "init", CreateMode.PERSISTENT);
        }

        // 注册监听者
        notifyServices = configurableListableBeanFactory.getBeansOfType(NotifyService.class).values();
        for(NotifyService notifyService : notifyServices) {
            addObserver(notifyService);
        }
        executor = configurableListableBeanFactory.getBean("executor", ThreadPoolTaskExecutor.class);

        zkClient.subscribeDataChanges(ApplicationMonitorRegister.STATE_PATH, new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                ServerEventInfo eventObject = SerializeUtil.unSerialize((byte[]) data);
                if(eventObject == null) {
                    return;
                }
                executor.execute(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                // 通知监所有听者
                                setChanged();
                                notifyObservers(eventObject);
                            } catch (Throwable e) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                });
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            zkClient.create(dataPath, "recreate", CreateMode.PERSISTENT);
                        } catch (Throwable e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                });
            }
        });
    }

}
